package org.ws.task.example.mq;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang.math.RandomUtils;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.ws.task.builder.TaskBuilder;
import org.ws.task.exception.ItemProcessException;
import org.ws.task.item.ItemWorkerProcessor;
import org.ws.task.item.PushTask;
import org.ws.task.model.TaskConfig.SendModel;
import org.ws.task.model.TaskItem;

import java.util.Collection;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Configuration
public class PushTaskConfig {

    @Bean
    public PushTask<String> uuidPushTask() {
        return TaskBuilder.builder("uuidPushTask")
                .config()
                .chunk(10)
                .fetchTimeout(10_000)
                .sendModel(SendModel.CLUSTERING)
                .leaderProcess(false)
                .<String>push()
                .processor(uuidItemProcessor())
                .builder();
    }

    @Bean
    public UuidItemProcessor uuidItemProcessor(){
        return new UuidItemProcessor();
    }

    public static class UuidItemProcessor implements ItemWorkerProcessor<String>{
        private AtomicInteger count = new AtomicInteger(0);

        @Override
        public void process(Collection<TaskItem<String>> items) throws ItemProcessException {
            if(CollectionUtils.isNotEmpty(items)){
                int i = count.addAndGet(items.size());
                if(i/100 % 10 == 0 || i >= 500000){
                    log.info("process count:{}",i);
                }
            }
        }

        public int getCount() {
            return count.get();
        }
    }
}
